package com.lingjtx.common.redis.stream;

import com.alibaba.fastjson2.JSON;
import com.lingjtx.common.core.util.StringUtil;
import com.lingjtx.common.thread.core.PoolNameConst;
import com.lingjtx.common.thread.core.ThreadPoolManager;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamCreateGroupArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.redisson.client.RedisBusyException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
public class RedisStreamConsumer implements ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(RedisStreamConsumer.class);

    private final RedissonClient redissonClient;
    private final ApplicationEventPublisher applicationEventPublisher;
    private final RedisEventListenerRegistry registry;
    private final ThreadPoolManager threadPoolManager;
    private ApplicationContext applicationContext;

    // 注册监听的类型（可动态扩展）
    private final List<Class<? extends RedisStreamTypedEvent>> eventTypes = new ArrayList<>();

    public RedisStreamConsumer(RedissonClient redissonClient,
                               ApplicationEventPublisher applicationEventPublisher,
                               RedisEventListenerRegistry registry,
                               ThreadPoolManager threadPoolManager) {
        this.redissonClient = redissonClient;
        this.applicationEventPublisher = applicationEventPublisher;
        this.registry = registry;
        this.threadPoolManager = threadPoolManager;
    }

    public void startConsumer() {
        // 获取所有实现了 RedisStreamTypedEvent 接口的类
        Map<String, RedisStreamTypedEvent> beans = applicationContext.getBeansOfType(RedisStreamTypedEvent.class);

        // 将实现类添加到 eventTypes 列表
        eventTypes.addAll(beans.values().stream()
                .map(RedisStreamTypedEvent::getClass)
                .toList());

        for (Class<? extends RedisStreamTypedEvent> type : eventTypes) {
            String streamKey = getStreamKey(type);
            String group = "consumer-group";
            String consumerName = StringUtil.getUuid();

            threadPoolManager.execute(PoolNameConst.REDIS_STREAM_CONSUMER, () -> {
                RStream<String, String> stream = redissonClient.getStream(streamKey);

                // 创建消费者组（幂等处理）
                try {
                    stream.createGroup(StreamCreateGroupArgs.name(group).id(StreamMessageId.NEWEST));
                } catch (RedisBusyException e) {
                    // 表示组已经存在，可以忽略
                    log.debug("Redis Stream Group [{}] already exists", group);
                }

                // 标记为可以退出循环的变量
                boolean running = true;
                int maxRetryCount = 5; // 最大消息消费数
                int retryCount = 0;

                while (running) {
                    if (retryCount >= maxRetryCount) {
                        running = false;  // 达到最大消息消费数
                        log.info("最大消息消费数，退出消费循环");
                    }

                    // 消费消息的逻辑...
                    retryCount++;

                    try {
                        Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
                                group,
                                consumerName,
                                StreamReadGroupArgs.neverDelivered().count(1).timeout(Duration.ofSeconds(5))
                        );

                        if (messages != null && !messages.isEmpty()) {
                            for (Map.Entry<StreamMessageId, Map<String, String>> entry : messages.entrySet()) {
                                StreamMessageId id = entry.getKey();
                                Map<String, String> fields = entry.getValue();
                                try {
                                    String payload = fields.get("payload");
                                    RedisStreamTypedEvent event = JSON.parseObject(payload, type);

                                    // 分发到 Spring 事件系统（暴露 Spring 事件分发）
                                    applicationEventPublisher.publishEvent(event);

                                    // 分发到自定义监听器（效率性能比 Spring 事件高）
                                    for (RedisEventListener<?> listener : registry.getListeners(type)) {
                                        invokeListener(listener, event);
                                    }

                                    // 手动 ack
                                    stream.ack(group, id);

                                } catch (Exception e) {
                                    log.error("Redis Stream 消费异常: ", e);
                                }
                            }
                        }
                    } catch (Exception e) {
                        log.error("Redis Stream 拉取消息失败: ", e);
                    }
                }
            });
        }
    }

    private String getStreamKey(Class<? extends RedisStreamTypedEvent> type) {
        try {
            return type.getDeclaredConstructor().newInstance().getType() + "-stream";
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @SuppressWarnings("unchecked")
    private <T extends RedisStreamTypedEvent> void invokeListener(RedisEventListener<?> listener, RedisStreamTypedEvent event) {
        ((RedisEventListener<T>) listener).onMessage((T) event);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
